package com.izhengyin.miaosha.service.impl;

import com.izhengyin.miaosha.service.QueuingService;
import org.springframework.data.redis.core.ReactiveStringRedisTemplate;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

@Service
public class RedisQueuingService implements QueuingService {
    private final String QUEUE_KEY_PREFIX = "QUEUE_";
    private final ReactiveStringRedisTemplate reactiveStringRedisTemplate;
    public RedisQueuingService(ReactiveStringRedisTemplate reactiveStringRedisTemplate) {
        this.reactiveStringRedisTemplate = reactiveStringRedisTemplate;
    }

    @Override
    public Mono<Boolean> enqueue(long itemId , String value) {
        final String key = getQueueKey(itemId);
        return this.reactiveStringRedisTemplate.opsForList()
                .leftPush(key,value)
                .map(v -> v > 0);
    }

    @Override
    public Flux<String> dequeue(long itemId, int size) {
        String queueKey = getQueueKey(itemId);
        return Flux.range(0,size)
                .flatMap(i -> this.reactiveStringRedisTemplate.opsForList().rightPop(queueKey));
    }

    private String getQueueKey(long itemId){
        return QUEUE_KEY_PREFIX+itemId;
    }

}
